//package cn.lagou.spark.demo
//
//import org.apache.kafka.clients.consumer.ConsumerConfig
//import org.apache.kafka.common.serialization.StringDeserializer
//import org.apache.log4j.{Level, Logger}
//import org.apache.spark.SparkConf
//import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
//import org.apache.spark.streaming.{Seconds, StreamingContext}
//
//object Demo {
//  def main(args: Array[String]): Unit = {
//    Logger.getLogger("org").setLevel(Level.ERROR)
//    val conf: SparkConf = new SparkConf()
//      .setMaster("local[*]")
//      .setAppName(this.getClass.getCanonicalName)
//    val ssc = new StreamingContext(conf, Seconds(5))
//    ssc.sparkContext.setLogLevel("WARN")
//
//    // 定义kafka相关参数
//    val kafkaParams: Map[String, Object] = getKafkaConsumerParams()
//    val topics: Array[String] = Array("topicB")
//
//    KafkaUtils.createDirectStream(
//      ssc,
//      // 本地策略
//      // LocationStrategies.PreferConsistent：大多数情况下使用的策略
//      // 将Kafka分区均匀的分布在Spark集群的 Executor上
//      LocationStrategies.PreferConsistent,
//      // 消费策略
//      // 用来订阅一组固定topic
//      ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
//    )
//
//    ssc.start()
//    ssc.awaitTermination()
//  }
//
//  def getKafkaConsumerParameters(groupId: String): Map[String,
//    Object] = {
//    Map[String, Object](
//      //      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092,linux122:9092,linux123:9092",
//      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092",
//      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
//      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
//      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
//      ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest",
//      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
//    )
//  }
//
//}
